package com.biconn.bigdata.jstrom.question.and.answer.pology;

import com.biconn.bigdata.jstrom.question.and.answer.bolt.QuestionAndAnswerOntimeBolt;
import com.biconn.bigdata.jstrom.question.and.answer.spout.QuestionAndAnswerOntimeSpout;
import com.biconn.bigdata.jstrom.util.LogbackInit;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;

public class QuestionAndAnswerOntimeLocalTopology {

	public static void main(String[] args) throws InterruptedException {
		String toponame = "question-and-answer-ontime";
		LogbackInit.initLogback("logback.xml");
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("out-word-ontime", new QuestionAndAnswerOntimeSpout(),1);
		builder.setBolt("word-count-ontime", new QuestionAndAnswerOntimeBolt(),1).shuffleGrouping("out-word-ontime");
		Config conf = new Config();
	    conf.setNumWorkers(1);
	    conf.setDebug(true);
	    conf.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 1);
	    if(args[0].equals("0")) {
	    	//本地模式:本地提交
			LocalCluster cluster = new LocalCluster();  
	        cluster.submitTopology("firstTopo", conf, builder.createTopology());  
	        //一定要等待足够的时间，否则程序没来得及运行就已经结束了，程序启动需要消耗时间
	        Thread.sleep(Integer.MAX_VALUE);
	   }else {
	    	//远程模式
	    	try {
				StormSubmitter.submitTopologyWithProgressBar(toponame, conf, builder.createTopology());
			} catch (AlreadyAliveException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (InvalidTopologyException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	    }
	}
	
	
	

}
